package com.hexinfo.dmpro.sparing.service.impl;


import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.hexinfo.dmpro.common.model.ClusterSparingConf;
import com.hexinfo.dmpro.common.model.MessageCenter;
import com.hexinfo.dmpro.common.service.MessageCenterService;
import com.hexinfo.dmpro.common.utils.CommonConstants;
import com.hexinfo.dmpro.common.utils.ScanCommonConstants;
import com.hexinfo.dmpro.sparing.async.AsyncFailTableRetrySource;
import com.hexinfo.dmpro.sparing.dto.TableSyncConfDTO;
import com.hexinfo.dmpro.sparing.model.HotStandbyLogSource;
import com.hexinfo.dmpro.sparing.model.HotStandbyLogTable;
import com.hexinfo.dmpro.sparing.model.SyncCompleteLibrary;
import com.hexinfo.dmpro.sparing.service.*;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Future;

@Service
@Slf4j
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class FailTableRetrySourceServiceImpl implements FailTableRetrySourceService {

    private final HotStandbyLogTableService hotStandbyLogTableService;
    private final HotStandbyLogSourceService hotStandbyLogSourceService;
    private final TableSyncMethodService tableSyncMethodService;
    private final AsyncFailTableRetrySource asyncFailTableRetrySource;
    private final ScanMetadataService scanMetadataService;
    private final MessageCenterService messageCenterService;
    private final WarnService warnService;

    /**
     * 异步线程超时时间
     */
    @Value("${threadMonitoring}")
    private Integer threadMonitoring;


    /**
     * 源级热备，失败表重新热备（手动重跑）
     */
    @Override
    public Boolean failTableRetryTable(List<HotStandbyLogTable> hotStandbyLogTables){
        //获取失败表所属任务
        LambdaQueryWrapper<HotStandbyLogSource> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(StrUtil.isNotBlank(hotStandbyLogTables.get(0).getSourceId()),HotStandbyLogSource::getId,hotStandbyLogTables.get(0).getSourceId())
                .eq(HotStandbyLogSource::getDel, CommonConstants.STATUS_NORMAL);
        HotStandbyLogSource hotStandbyLogSource = hotStandbyLogSourceService.getBaseMapper().selectOne(wrapper);
        //修改源热备任务重跑状态（重跑中）
        hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPZ.value);
        hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.CPZ.value);
        hotStandbyLogSourceService.updateById(hotStandbyLogSource);
        //获取基础配置
        TableSyncConfDTO tableSyncConfDTO = gainConfig(hotStandbyLogSource.getSyncRangeStartTime());

        //遍历失败表，执行表重跑
        Boolean aBoolean = tableHotStandbySync(hotStandbyLogTables, tableSyncConfDTO);

        //获取失败表集合
        List<HotStandbyLogTable> hotStandbyLogTables1 = hotStandbyLogTableService.queryLikeList(hotStandbyLogSource.getId(), ScanCommonConstants.ClusterName.SB.value);
        if (hotStandbyLogTables1 == null || hotStandbyLogTables1.isEmpty()){
            hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPCG.value);
            hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            countNumber(hotStandbyLogSource);
            //修改消息中心数据
            MessageCenter mess = messageCenterService.getById(hotStandbyLogSource.getMessageId());
            mess.setDataType(ScanCommonConstants.ClusterName.YWC.value);
            messageCenterService.updateById(mess);
        }else {
            hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPSB.value);
            hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            //钉钉告警
            warnService.warnSourceFail(hotStandbyLogSource.getId());
        }
        hotStandbyLogSourceService.updateById(hotStandbyLogSource);
        return aBoolean;
    }


    /**
     * 源级热备，失败表重新热备（自动失败重跑）
     * @param sourceName 源名
     */
    @Override
    public Boolean failTableRetrySource(String sourceName,HotStandbyLogSource hotStandbyLogSource){
        //获取失败表集合
        List<HotStandbyLogTable> hotStandbyLogTables = hotStandbyLogTableService.queryLikeList(hotStandbyLogSource.getId(), ScanCommonConstants.ClusterName.SB.value);
        if (hotStandbyLogTables == null || hotStandbyLogTables.isEmpty()){
            log.warn(sourceName+"没有失败表，不进行重跑！！！");
            return true;
        }
        //修改源热备任务重跑状态（重跑中）
        hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPZ.value);
        hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.CPZ.value);
        hotStandbyLogSourceService.updateById(hotStandbyLogSource);
        //获取基础配置
        TableSyncConfDTO tableSyncConfDTO = gainConfig(hotStandbyLogSource.getSyncRangeStartTime());

        //遍历失败表，执行表重跑
        Boolean aBoolean = tableHotStandbySync(hotStandbyLogTables, tableSyncConfDTO);

        //修改源热备任务结束重跑状态
        if (aBoolean){
            hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPCG.value);
            hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.YWC.value);
            countNumber(hotStandbyLogSource);
            //修改消息中心数据
            MessageCenter mess = messageCenterService.getById(hotStandbyLogSource.getMessageId());
            mess.setDataType(ScanCommonConstants.ClusterName.YWC.value);
            messageCenterService.updateById(mess);
        }else {
            hotStandbyLogSource.setRerunState(ScanCommonConstants.ClusterName.CPSB.value);
            hotStandbyLogSource.setTaskState(ScanCommonConstants.ClusterName.SB.value);
            //钉钉告警
            warnService.warnSourceFail(hotStandbyLogSource.getId());
        }
        hotStandbyLogSourceService.updateById(hotStandbyLogSource);
        return aBoolean;
    }

    /**
     * 遍历失败表，执行表重跑（串行）
     */
    public Boolean tableHotStandbySync(List<HotStandbyLogTable> hotStandbyLogTables,TableSyncConfDTO tableSyncConfDTO){
        Boolean aBoolean = true;
        for (HotStandbyLogTable hotStandbyLogTable : hotStandbyLogTables) {
            Boolean b = asyncFailTableRetrySource.tableHotStandbySync(hotStandbyLogTable, tableSyncConfDTO);
            //修改表结束重跑状态
            if (!b){
                aBoolean = false;
            }
        }
        return aBoolean;
    }

    /**
     * 遍历失败表，执行表重跑（异步）
     */
    public Boolean tableHotStandby(List<HotStandbyLogTable> hotStandbyLogTables,TableSyncConfDTO tableSyncConfDTO){
        List<Future<Boolean>> futures = new ArrayList<>();
        for (HotStandbyLogTable hotStandbyLogTable : hotStandbyLogTables) {
            futures.add(asyncFailTableRetrySource.tableHotStandby(hotStandbyLogTable,tableSyncConfDTO));
        }
        //判断所有获取是否全部完成
        Boolean aBoolean = tableSyncMethodService.futureStatus(futures, threadMonitoring);
        return aBoolean;
    }

    /**
     * 查询日志ID下全部的表信息做源总量和目标总量计算
     */
    public void countNumber(HotStandbyLogSource hotStandbyLogSource){
        Long numSource =  0L;
        Long numTarget =  0L;
        List<HotStandbyLogTable> tableList = hotStandbyLogTableService.querySourceId(hotStandbyLogSource.getId());
        if(ObjectUtil.isNotEmpty(tableList)){
            for (HotStandbyLogTable table: tableList) {
                if(ObjectUtil.isNotEmpty(table.getSourceCountNumber())){
                    numSource += table.getSourceCountNumber();
                }
                if(ObjectUtil.isNotEmpty(table.getTargetCountNumber())){
                    numTarget += table.getTargetCountNumber();
                }
            }
        }
        hotStandbyLogSource.setSourceCountNumber(numSource);
        hotStandbyLogSource.setTargetCountNumber(numTarget);
    }

    /**
     * 获取基础配置
     */
    public TableSyncConfDTO gainConfig(String rangeStartDate){
        //查询集群信息
        LambdaQueryWrapper<ClusterSparingConf> jQWrapper = Wrappers.lambdaQuery();
        LambdaQueryWrapper<ClusterSparingConf> wGWrapper = Wrappers.lambdaQuery();
        ClusterSparingConf jQCluster = scanMetadataService.wrapperClusterSparingConf(jQWrapper,"金桥");
        ClusterSparingConf wGCluster = scanMetadataService.wrapperClusterSparingConf(wGWrapper,"万国");
        TableSyncConfDTO tableSyncConfDTO =new TableSyncConfDTO()
                .setJQHdfsPrefix(jQCluster.getHdfsPrefixPath())
                .setWGHdfsPrefix(wGCluster.getHdfsPrefixPath())
                .setRangeStartDate(rangeStartDate);
        return tableSyncConfDTO;
    }

}
